package org.leiyang.netty.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.internal.StringUtil;
import org.leiyang.common.etcd.IConfigCenter;
import org.leiyang.common.holder.OnlineServerHolder;
import org.leiyang.common.redis.ReactiveRedisCache;
import org.leiyang.common.util.AsyncPool;
import org.leiyang.common.util.CpuNum;
import org.leiyang.common.util.IpUtils;
import org.leiyang.netty.listener.IClientChangeListener;
import org.leiyang.netty.listener.INettyMsgFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static org.leiyang.common.util.Constants.*;

@Component
@Order(value = 100)
public class MsgServer implements ApplicationRunner, ApplicationListener<ContextClosedEvent> {
    private final Logger logger = LoggerFactory.getLogger(getClass());
    private final static ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

    @Value("${netty.port:18080}")
    private int port;

    @Value("${local.address}")
    private String localAddress;

    /**
     * 当前应用实例的端口
     */
    @Value("${server.port}")
    private int webPort;

    /**
     * 当前应用实例的RSocket服务端端口
     */
    @Value("${spring.rsocket.server.port}")
    private int rsocketPort;
    /**
     * 当前应用实例的RSocket服务的根路径
     */
    @Value("${spring.rsocket.server.mapping-path}")
    private String rsocketPath;

    /**
     * 是否继续上报注册
     */
    private volatile boolean canUpload = true;

    /**
     * 本机IP地址
     */
    private static String localIp;

    private final IClientChangeListener clientChangeListener;
    private final List<INettyMsgFilter> messageFilters;
    private final OnlineServerHolder onlineServerHolder;
    private final IConfigCenter configCenter;
    private final ReactiveRedisCache redisCache;
    public MsgServer(IClientChangeListener clientChangeListener, List<INettyMsgFilter> messageFilters, OnlineServerHolder onlineServerHolder,
                     IConfigCenter configCenter, ReactiveRedisCache redisCache) {
        this.clientChangeListener = clientChangeListener;
        this.messageFilters = messageFilters;
        this.onlineServerHolder = onlineServerHolder;
        this.configCenter = configCenter;
        this.redisCache = redisCache;
    }

    public void startMsgServer(int port) throws Exception {
        //boss单线程
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup(CpuNum.workerCount());
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    //保持长连接
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    //出来网络io事件，如记录日志、对消息编解码等
                    .childHandler(new ChildChannelHandler());
            //绑定端口，同步等待成功
            ChannelFuture future = bootstrap.bind(port).sync();
            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                bossGroup.shutdownGracefully (1000, 3000, TimeUnit.MILLISECONDS);
                workerGroup.shutdownGracefully (1000, 3000, TimeUnit.MILLISECONDS);
            }));
            //缓存启动的msgServer信息
            localIp = gotLocalIp();
            String msgPort = String.valueOf(port);
            onlineServerHolder.joinOnlineServer(localIp, msgPort).subscribe();
            //缓存对应的WEB通信地址信息(tomcat和RSocket)
            redisCache.setCacheObject(MSG_SER_WEB_INFO.concat(localIp).concat(":").concat(msgPort), localIp.concat(":").concat(String.valueOf(webPort)))
                    .flatMap(setWeb -> redisCache.setCacheObject(MSG_SER_RSOCKET_INFO.concat(localIp).concat(":").concat(msgPort), localIp.concat(":")
                            .concat(String.valueOf(rsocketPort))
                            .concat(rsocketPath)
                        )
                    )
                    .subscribe(re -> logger.info("当前MsgServer的WEB通信地址已缓存."));
            logger.info("MsgServer start success,please open your web browser " +
                    "and navigate to http://localhost:{}/websocket", port);
            //向ETCD注册当前MsgServer
            registerMsgServer();
            //等待服务器监听端口关闭（执行到此处会等待，即不会往下执行后续的代码）
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            logger.error("MsgServer 启动失败，", e);
        } finally {
            //优雅退出，释放线程池资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        logger.info("MsgServer开始启动...");
        startMsgServer(port);
    }

    @Override
    public void onApplicationEvent(ContextClosedEvent event) {
        if(!scheduledExecutorService.isShutdown()) {
            scheduledExecutorService.shutdown();
        }
        AsyncPool.shutDown();
        String reLocalIp = StringUtil.isNullOrEmpty(localIp) ? gotLocalIp() : localIp;
        String msgPort = String.valueOf(port);
        /**
         * 此处与org.leiyang.netty.starters.EtcdStarter.monitorMsgServerOnline一起作为双重保障来确保MsgServer的在线信息的实时性
         */
        onlineServerHolder.removeOffLineServer(reLocalIp, msgPort)
                .flatMap(re -> redisCache.deleteKeys(MSG_SER_WEB_INFO.concat(localIp).concat(":").concat(msgPort),
                        MSG_SER_RSOCKET_INFO.concat(localIp).concat(":").concat(msgPort))
                )
                .subscribe(reNum -> logger.info("开始移除当前MsgServer[{}:{}]记录", localIp, port));
        logger.info("MsgServer[{}:{}]已完成执行下线操作....", localIp, msgPort);
    }

    /**
     * handler类
     */
    private class ChildChannelHandler extends ChannelInitializer<Channel> {
        @Override
        protected void initChannel(Channel ch) {
            logger.info("服务端开始执行initChannel操作---");
//            ByteBuf delimiter = Unpooled.copiedBuffer(Constants.DELIMITER.getBytes());
            ch.pipeline()
//                    .addLast(new DelimiterBasedFrameDecoder(Constants.MAX_LENGTH, delimiter))
//                    .addLast(new MsgDecoder())
//                    .addLast(new MsgEncoder())
                    //此处需要适配websocket协议
                    //添加压缩处理
                    .addLast(new WebSocketServerCompressionHandler())
                    //http协议处理(序列化和反序列化)
                    .addLast(new HttpServerCodec())
                    //添加aggregator，将多个消息转换为单一的HttpObject
                    .addLast(new HttpObjectAggregator(512 * 1024))
                    //添加WebSocket处理器
                    .addLast(new WebSocketServerProtocolHandler("/websocket", null, true))
                    .addLast(new MsgServerHandler(clientChangeListener, messageFilters, port, localAddress));
        }
    }

    /**
     * 获取本机IP
     * @return 本机IP
     */
    private String gotLocalIp() {
        return StringUtil.isNullOrEmpty(localAddress) ? IpUtils.getIp() : localAddress;
    }

    /**
     * 注册MsgServer
     */
    private void registerMsgServer() {
        //开启上传worker信息
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            try {
                if (canUpload) {
                    logger.debug("开始定时向ETCD注册当前MsgServer!");
                    uploadMsgServerSelfInfo();
                }
            } catch (Exception e) {
                logger.error("向ETCD里注册MsgServer失败!", e);
            }

        }, 0, 5, TimeUnit.SECONDS);
    }

    /**
     * 向ETCD里注册自己的信息
     * KEY格式: online_server_key + serverIp（.替换为_） + '&' + port
     */
    private void uploadMsgServerSelfInfo() {
//        String localIp = gotLocalIp();
        String key = online_server_key.concat(localIp.replaceAll("\\.", "_")).concat("&").concat(String.valueOf(port));
        String value = localIp.concat(":").concat(String.valueOf(port));
        configCenter.putAndGrant(key, value, 8);
    }
}
